package db

import (
	"context"
	"errors"
	"github.com/go-redis/redis/v8"
	"sync"
)

const (
	DB_REDIS_MSG = 1 //用户信息数据库
	//DB_REDIS_USER = 1003
	//DB_REDIS_USER = 1004

)

type MsgDbIns struct {
	ctx  context.Context
	rdb  *redis.Client
	pipe redis.Pipeliner
}

var msgDbIns *MsgDbIns
var MsgDBOnce sync.Once

// Instance 考虑到事务的操作便利，所以将所有的数据都放到序号0数据库中
func MsgInstance() *MsgDbIns {
	MsgDBOnce.Do(func() {
		msgDbIns = &MsgDbIns{
			ctx: context.Background(),
			rdb: redis.NewClient(&redis.Options{
				Addr:     ":6379",
				Password: "",           // no password set
				DB:       DB_REDIS_MSG, // use default DB
			}),
		}
	})
	return msgDbIns
}

// Ping 检查Redis通讯
func (ins *MsgDbIns) Ping() error {
	return ins.rdb.Ping(ins.ctx).Err()
}

// StartPipeline 开启事务
//注意，开启事务之后所有的操作就必须使用pipe来实现
func (ins *MsgDbIns) StartPipeline() {
	ins.pipe = ins.rdb.TxPipeline()
}

// Exec 考虑到事务的执行是一次出全部的结果，所以如果有要取出结果的操作就不建议使用事务了
// 譬如在保存mark的最后需要保存mark，position，user但均不需要结果
// 这里算是刻意去避开了对返回结果的解析
func (ins *MsgDbIns) Exec() (err error) {
	if ins.pipe == nil {
		err = errors.New("事务未开启。")
		return err
	}
	_, err = ins.pipe.Exec(ins.ctx)
	//对事务执行完成之后，就需要将pipe进行清空，避免影响后面的执行
	ins.pipe = nil
	return err
}

// SetValue 写入KEY-VALUE
func (ins *MsgDbIns) SetValue(k string, v interface{}) error {
	if ins.pipe != nil {
		return ins.pipe.Set(ins.ctx, k, v, 0).Err()
	}
	return ins.rdb.Set(ins.ctx, k, v, 0).Err()
}

// GetValue 读取VALUE
func (ins *MsgDbIns) GetValue(k string) (string, error) {
	if ins.pipe != nil {
		return ins.pipe.Get(ins.ctx, k).Result()
	}
	return ins.rdb.Get(ins.ctx, k).Result()
}

// AddMsg 消息入栈
// 消息入栈之后会返回一个id暂时不用
func (ins *MsgDbIns) AddMsg(msgKey string, field string, value string) (string, error) {
	if ins.pipe != nil {
		return ins.pipe.XAdd(ins.ctx, &redis.XAddArgs{
			Stream: msgKey,
			ID:     "*",
			Values: []string{field, value},
		}).Result()
	}
	return ins.rdb.XAdd(ins.ctx, &redis.XAddArgs{
		Stream: msgKey,
		ID:     "*",
		Values: []string{field, value},
	}).Result()
}

// ReadMsg 读取最新消息
// 读取消息之后会获取到id作为下次读取的起始点
func (ins *MsgDbIns) ReadMsg(streams []string) ([]redis.XStream, error) {
	if ins.pipe != nil {
		return ins.pipe.XRead(ins.ctx, &redis.XReadArgs{
			Block:   -1,
			Streams: streams,
		}).Result()
	}
	return ins.rdb.XRead(ins.ctx, &redis.XReadArgs{
		Block:   -1,
		Streams: streams,
	}).Result()
}

// RangeMsg 读取消息
// 这里用作读取历史消息
func (ins *MsgDbIns) RangeMsg(stream, start, stop string, count int64) ([]redis.XMessage, error) {
	if ins.pipe != nil {
		return ins.pipe.XRevRangeN(ins.ctx, stream, start, stop, count).Result()
	}
	return ins.rdb.XRevRangeN(ins.ctx, stream, start, stop, count).Result()
}

// XLen 读取消息长度
// 这里用作读取历史消息长度
func (ins *MsgDbIns) XLen(stream string) (int64, error) {
	if ins.pipe != nil {
		return ins.pipe.XLen(ins.ctx, stream).Result()
	}
	return ins.rdb.XLen(ins.ctx, stream).Result()
}

// Sadd 添加集合
func (ins *MsgDbIns) Sadd(k string, v string) (int64, error) {
	if ins.pipe != nil {
		return ins.pipe.SAdd(ins.ctx, k, v).Result()
	}
	return ins.rdb.SAdd(ins.ctx, k, v).Result()
}

//Exist 集合是否包含 成员
func (ins *MsgDbIns) Exist(k, member string) (bool,error) {
	return ins.rdb.SIsMember(ins.ctx,k, member).Result()
}
